package com.im.service.websocket.handler;

import com.alibaba.fastjson2.JSON;
import com.im.service.common.constant.ImConstant;
import com.im.service.infra.entity.ChatHistory;
import com.im.service.infra.entity.ImUser;
import com.im.service.exception.BusinessException;
import com.im.service.exception.ErrorCode;
import com.im.service.service.ChatHistoryService;
import com.im.service.service.ImUserService;
import com.im.service.websocket.message.MessageProcessorFactory;
import com.im.service.websocket.message.dto.SentMessage;
import com.im.service.websocket.message.enums.MessageStatus;
import com.im.service.websocket.session.SessionManager;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.WebSocketSession;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.time.LocalDateTime;
import java.util.Map;
import java.util.Scanner;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

@Component
@Slf4j
@RequiredArgsConstructor
public class ChatWebSocketHandler implements WebSocketHandler {
    // 存储所有用户会话
    private final ImUserService imUserService;
    private final ChatHistoryService chatHistoryService;
    private final MessageProcessorFactory messageProcessorFactory;

    @Override
    public Mono<Void> handle(WebSocketSession session) {
        // 用户名作为查询参数传递，用户登录后被分配唯一 ID
        String userId = session.getHandshakeInfo().getUri().getQuery();
        // 首先检查用户是否存在
        return this.rejectNotRegistryUser(userId) // 用户校验
                .flatMap(this::updateUserOnlineTime) // 更新用户上线时间
                .then(Mono.fromRunnable(() -> SessionManager.putSession(userId, session))) // 管理socket
                .then(batchSendOfflineMessage(userId)) // 发送离线消息
                .then(Mono.defer(() -> {
                    // 处理从客户端接收到的消息
                    Mono<Void> receive = session.receive()
                            .map(WebSocketMessage::getPayloadAsText)
                            .flatMap(messageProcessorFactory.switchAndHandlerMessage(userId))
                            .then();

                    return receive
                            .doFinally(signalType -> SessionManager.removeSession(userId)) // 移除已断开的会话
                            .then();
                }));
    }

    private Mono<ImUser> rejectNotRegistryUser(String userId) {
        return imUserService.findByUserId(userId)  // 返回一个 Mono<User> 或 Mono.empty()
                .switchIfEmpty(Mono.defer(() -> {
                    log.error("未找到用户【{}】，拒绝连接！", userId);
                    return Mono.error(new BusinessException(ErrorCode.USER_NOT_REGISTRY_ERROR)); // 如果返回的是 Mono.empty()，抛出异常
                }));
    }

    private Mono<Void> updateUserOnlineTime(ImUser imUser) {
        imUser.setLastOnlineTime(LocalDateTime.now());
        log.info("用户[{}]已上线", imUser.getId());
        return imUserService.saveOrUpdate(imUser).then();
    }

    private Mono<Void> batchSendOfflineMessage(String receiverId) {
        return chatHistoryService.queryOfflineMessageByReceiverId(receiverId)
                .flatMap(message -> SessionManager.getSender(receiverId).sendMessage(message))
                .filter(message -> message.getForwardSuccessTime() != null)
                .flatMap(chatHistoryService::saveChatHistory)
                .then();
    }
}
